package com.navi.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.navi.entity.ColumnEntity;
import com.navi.entity.HisConfigure;
import com.navi.entity.TableEntity;
import com.navi.exception.HmsException;
import com.navi.util.JacksonUtil;
import com.navi.util.LogUtil;
import com.navi.util.TableJdbcUtil;
import com.navi.util.TestConnectionUtil;
import com.rabbitmq.client.*;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.io.IOException;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeoutException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * Created by admin on 2017/11/28.
 */
@Service
@Slf4j
public class JdbcLoaderService {

    @Value("${spring.rabbitmq.exchange}")
    private String exchange;

    @Value("${spring.rabbitmq.error.queue.routekey}")
    private String errorQueueRouteKey;

    @Value("${spring.rabbitmq.error.queue.name}")
    private String errorQueueName;


    @Value("${spring.rabbitmq.error.queue.routekey2}")
    private String errorQueueRouteKey2;

    @Value("${spring.rabbitmq.error.queue.name2}")
    private String errorQueueName2;


    @Autowired
    private ConnectionFactory connectionFactory;

    @Autowired
    private TestConnectionUtil testConnectionUtil;

    @Autowired
    private JedisPool jedisPool;

    public void subMainProc(String queueName) throws IOException, TimeoutException {

        //获取jsonList

        Channel channel = null;
        Connection connection = null;
        try {
            connection = connectionFactory.newConnection();
//                logger.info("connected to RabbitMQ server");

            //connect to exchange
            channel = connection.createChannel();
            channel.exchangeDeclare(exchange, ExchangeTypes.TOPIC, true);

            final List<JSONObject> jsonObjectList = new ArrayList<>();

            while (true) {

                final GetResponse getResponse = channel.basicGet(queueName, true);
                if (getResponse == null) {
                    break;
                }

                String message = new String(getResponse.getBody(), "utf-8");
                log.info("message from [{}] , {} ", queueName, message);


                if (StringUtils.startsWith(message, "{")) {
                    jsonObjectList.add((JSONObject) JSONObject.parse(message));
                } else if (StringUtils.startsWith(message, "[")) {
                    jsonObjectList.addAll(JSONArray.parseArray(message).stream().map(obj -> (JSONObject) obj).collect(Collectors.toList()));
                }


                if (jsonObjectList.isEmpty()) {
                    log.info("no message in queue: {}", queueName);
                    return;
                }

                jsonObjectList.forEach(jsonObject -> {
                    /**
                     * 获取表名
                     */
                    final String tableName = jsonObject.getString("ope_tbl_name");
                    if (StringUtils.isEmpty(tableName)) {
                        log.error("消息中没有ope_tbl_name结点,错误");
                        throw new HmsException("消息中没有ope_tbl_name结点,错误");
                    }
                    /**
                     * 获取配置
                     */
                    try (Jedis jedis = jedisPool.getResource()) {

                        final String strHisconfig = jedis.hget("hisconfig", tableName);

                        //如果找不到对应的配置文件，那么直接送ERROR_QUEUE
                        if (StringUtils.isEmpty(strHisconfig)) {
                            log.error("table {} can not find config file", tableName);
                            if (StringUtils.equalsIgnoreCase(errorQueueName, queueName)) {
                                sendJsonToErrorQueue(jsonObject.toJSONString(), errorQueueName2);
                            } else {
                                sendJsonToErrorQueue(jsonObject.toJSONString());
                            }

                            return;
                        }
                        HisConfigure hisConfigure = JacksonUtil.fromJson(strHisconfig, HisConfigure.class);

                        //如果配置文件不是正确的格式，也直接送ERROR_QUEUE
                        if (hisConfigure == null) {
                            log.error("table {}  config can not parse correctly", tableName);
                            if (StringUtils.equalsIgnoreCase(errorQueueName, queueName)) {
                                sendJsonToErrorQueue(jsonObject.toJSONString(), errorQueueName2);
                            } else {
                                sendJsonToErrorQueue(jsonObject.toJSONString());
                            }
                            return;
                        }

                        try (final java.sql.Connection dbConnection = testConnectionUtil.getDefaultConnection()) {

                            if (hisConfigure.getPptFlg()) {
                                final String pptTableSchema = hisConfigure.getPptTableSchema();
                                final String pptTableName = hisConfigure.getPptTableName();
                                mergeProc(dbConnection, pptTableSchema, pptTableName, jsonObject);
                                log.info("merge successful");
                            }

                            if (hisConfigure.getHisFlg()) {
                                final String hisTableSchema = hisConfigure.getHisTableSchema();
                                final String hisTableName = hisConfigure.getHisTableName();
                                insertProc(dbConnection, hisTableSchema, hisTableName, jsonObject);
                                log.info("insert successful");
                            }
                        } catch (SQLException e) {
                            e.printStackTrace();
                            throw new HmsException(e);
                        } catch (ClassNotFoundException e) {
                            e.printStackTrace();
                            throw new HmsException(e);
                        }
                    } catch (Exception e) {
                        LogUtil.pringStackTrace(e);
                        try {
                            if (StringUtils.equalsIgnoreCase(errorQueueName, queueName)) {
                                sendJsonToErrorQueue(jsonObject.toJSONString(), errorQueueName2);
                            } else {
                                sendJsonToErrorQueue(jsonObject.toJSONString());
                            }
                        } catch (IOException e1) {
                            LogUtil.pringStackTrace(e);
                            e1.printStackTrace();
                        } catch (TimeoutException e1) {
                            LogUtil.pringStackTrace(e);
                            e1.printStackTrace();
                        }
                    }


                    //如果报错，送ErrorQueue
                });


            }
        } catch (Exception ex) {
            ex.printStackTrace();

            log.error(ex.toString());
        } finally {
            if (connection != null) {
                connection.close();
            }
        }


    }

    private void insertProc(java.sql.Connection dbConnection, String hisTableSchema, String hisTableName, JSONObject jsonObject) throws SQLException {
        /**
         * 获取表结构
         */
        Statement statement = dbConnection.createStatement();

        final TableEntity tableEntity = TableJdbcUtil.getTableMetaInfo(hisTableSchema, hisTableName, statement);

        //拼sql
        final List<ColumnEntity> columnEntityList = null;
        tableEntity.getColumnEntityList();

        final List<String> columnNameList = columnEntityList.stream().map(columnEntity -> columnEntity.getColumnName()).collect(Collectors.toList());

        StringBuffer sql = buildJdbcInsertSql(hisTableSchema, hisTableName, columnNameList);


        //创建PreparedStatement

        executeSql(dbConnection, jsonObject, columnEntityList, sql);
    }

    private void executeSql(java.sql.Connection dbConnection, JSONObject jsonObject, List<ColumnEntity> columnEntityList, StringBuffer sql) throws SQLException {
        final PreparedStatement preparedStatement = dbConnection.prepareStatement(sql.toString());

        //给PreparedStatement赋值

        for (int parameterIndex = 1; parameterIndex <= columnEntityList.size(); parameterIndex++) {
            final ParameterMetaData parameterMetaData;
            try {
                parameterMetaData = preparedStatement.getParameterMetaData();
                final int parameterType = parameterMetaData.getParameterType(parameterIndex);
                final String columnName = columnEntityList.get(parameterIndex - 1).getColumnName();
                String value = getJsonValue(jsonObject, columnName);
                preparedStatement.setObject(parameterIndex, value, parameterType);
            } catch (SQLException e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }


        log.info(preparedStatement.toString());
        //执行
        preparedStatement.execute();

        log.info("sql execute successful");
    }

    private StringBuffer buildJdbcInsertSql(String hisTableSchema, String hisTableName, List<String> columnNameList) {
        StringBuffer sql = new StringBuffer("INSERT INTO ");
        sql.append(hisTableSchema).append(".").append(hisTableName);
        sql.append("(").append(columnNameList.stream().collect(Collectors.joining(","))).append(")");
        sql.append("VALUES(").append(columnNameList.stream().map(c -> "?").collect(Collectors.joining(","))).append(")");
        return sql;
    }

    private void mergeProc(java.sql.Connection dbConnection, String pptTableSchema, String pptTableName, JSONObject jsonObject) throws SQLException {

        /**
         * 获取表结构
         */
        Statement statement = dbConnection.createStatement();

        final TableEntity tableEntity = TableJdbcUtil.getTableMetaInfo(pptTableSchema, pptTableName, statement);

        //拼sql
        List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();

        final List<String> columnNameList = columnEntityList.stream().map(columnEntity -> columnEntity.getColumnName()).collect(Collectors.toList());

        StringBuffer sql = null;

        //查询数据库，如果没有insert，如果有update
        if (!tableEntity.getHasPrimaryKey()) {
            sql = buildJdbcInsertSql(pptTableSchema, pptTableName, columnNameList);
        } else {

            //获取主键
            final List<ColumnEntity> primaryColumnList = columnEntityList.stream()
                    .filter(columnEntity -> columnEntity.getPrimaryKeyFlg()).collect(Collectors.toList());

            boolean isExisted = false;
            //拼根据主键查询的SQL
            StringBuffer querySql = new StringBuffer("SELECT 1 FROM ");
            querySql.append(pptTableSchema).append(".").append(pptTableName);
            querySql.append(" WHERE ");
            querySql.append(primaryColumnList.stream().map(columnEntity -> columnEntity.getColumnName() + "= ? ").collect(Collectors.joining(" AND ")));

            final PreparedStatement preparedStatement = dbConnection.prepareStatement(querySql.toString());

            //给PreparedStatement赋值

            for (int parameterIndex = 1; parameterIndex <= primaryColumnList.size(); parameterIndex++) {
                final ParameterMetaData parameterMetaData;
                try {
                    parameterMetaData = preparedStatement.getParameterMetaData();
                    final int parameterType = parameterMetaData.getParameterType(parameterIndex);
                    final String columnName = primaryColumnList.get(parameterIndex - 1).getColumnName();
                    String value = getJsonValue(jsonObject, columnName);
                    preparedStatement.setObject(parameterIndex, value, parameterType);
                } catch (SQLException e) {
                    e.printStackTrace();
                    throw new RuntimeException(e);
                }
            }

            final ResultSet resultSet = preparedStatement.executeQuery();
            if (resultSet.next()) {
                isExisted = true;
            }


            if (!isExisted) {
                sql = buildJdbcInsertSql(pptTableSchema, pptTableName, columnNameList);
            } else {
                final List<ColumnEntity> normalColumnEntityList = columnEntityList.stream().filter(c -> !c.getPrimaryKeyFlg()).collect(Collectors.toList());
                final List<String> normalColumnNameList = normalColumnEntityList.stream().map(c -> c.getColumnName()).collect(Collectors.toList());


                final List<String> primaryColumnNameList = primaryColumnList.stream().map(ColumnEntity::getColumnName).collect(Collectors.toList());
                sql = buildJdbcUpdateSql(pptTableSchema, pptTableName, normalColumnNameList, primaryColumnNameList);

                normalColumnEntityList.addAll(primaryColumnList);
                columnEntityList = normalColumnEntityList;
            }
        }


        //创建PreparedStatement

        executeSql(dbConnection, jsonObject, columnEntityList, sql);
    }

    private StringBuffer buildJdbcUpdateSql(String tableSchema, String tableName, List<String> normalColumnNameList, List<String> primaryColumnNameList) {
        StringBuffer sql = new StringBuffer("UPDATE ");
        sql.append(tableSchema).append(".").append(tableName);
        sql.append(" SET ").append(normalColumnNameList.stream().map(column -> column + " = ?").collect(Collectors.joining(",")));
        sql.append(" WHERE ").append(primaryColumnNameList.stream().map(column -> column + " = ?").collect(Collectors.joining(" AND ")));
        return sql;
    }

    private String getJsonValue(JSONObject jsonObject, String columnName) {
        String value = jsonObject.getString(columnName);

        if (StringUtils.isEmpty(value)) {
            String colCamelName = underline2Camel(columnName, true);
            value = jsonObject.getString(colCamelName);
        }

        return StringUtils.isEmpty(value) ? null : value;
    }

    public static String underline2Camel(String line, boolean smallCamel) {
        if (StringUtils.isEmpty(line)) {
            return "";
        }
        StringBuffer sb = new StringBuffer();
        Pattern pattern = Pattern.compile("([A-Za-z\\d]+)(_)?");
        Matcher matcher = pattern.matcher(line);
        while (matcher.find()) {
            String word = matcher.group();
            sb.append(smallCamel && matcher.start() == 0 ? Character.toLowerCase(word.charAt(0)) : Character.toUpperCase(word.charAt(0)));
            int index = word.lastIndexOf('_');
            if (index > 0) {
                sb.append(word.substring(1, index).toLowerCase());
            } else {
                sb.append(word.substring(1).toLowerCase());
            }
        }
        return sb.toString();
    }

    public void sendJsonToErrorQueue(String message) throws IOException, TimeoutException {
        sendJsonToErrorQueue(message, errorQueueRouteKey);
    }

    public void sendJsonToErrorQueue(String message, String routeKey) throws IOException, TimeoutException {
        try (com.rabbitmq.client.Connection connection = connectionFactory.newConnection()) {
            final Channel channel = connection.createChannel();
            channel.basicPublish(exchange, routeKey,
                    new AMQP.BasicProperties.Builder()
                            .contentType("text/plain")
                            .deliveryMode(2)//设置消息持久性:2永不过期
                            .priority(1).build(), message.getBytes());
            log.error("errors ,send to errorQueue");
//            list.forEach(message -> sendMessage(routeKey, channel, message));
        }
    }
}
